package com.seezoon.application.event;

import com.seezoon.domain.service.DeviceInfoService;
import com.seezoon.domain.valueobj.PingVO;
import com.seezoon.infrastructure.constants.Constants;
import com.seezoon.infrastructure.properties.AppProperties;
import com.seezoon.infrastructure.properties.DeviceProperties;
import java.util.HashSet;
import java.util.Queue;
import java.util.Set;
import java.util.concurrent.ConcurrentLinkedQueue;
import lombok.extern.slf4j.Slf4j;
import org.springframework.context.ApplicationListener;
import org.springframework.context.PayloadApplicationEvent;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

@Slf4j
@Component
public class PingEventListener implements ApplicationListener<PayloadApplicationEvent<PingVO>> {

    private final DeviceInfoService deviceInfoService;
    private final Queue<PingVO> heartbeatQueue = new ConcurrentLinkedQueue<>();
    private final int batchSize;
    private final int maxQueueSize;

    public PingEventListener(DeviceInfoService deviceInfoService, AppProperties appProperties) {
        this.deviceInfoService = deviceInfoService;
        DeviceProperties device = appProperties.getDevice();
        this.batchSize = device.getBatchSize();
        this.maxQueueSize = device.getMaxQueueSize();
    }

    // 定时器，每隔一定时间刷一次
    @Scheduled(fixedDelayString = Constants.BATCH_SUBMIT_INTERVAL)
    public void scheduledFlush() {
        if (!heartbeatQueue.isEmpty()) {
            try {
                flush();
            } catch (Throwable e) {
                log.error("batch flush ping error", e);
            }
        }
    }

    @Override
    public void onApplicationEvent(PayloadApplicationEvent<PingVO> event) {
        PingVO payload = event.getPayload();
        heartbeatQueue.add(payload);
        int size = heartbeatQueue.size();
        if (size >= maxQueueSize) {
            heartbeatQueue.clear();
            log.warn("heartbeatQueue size:{} maxQueueSize limit:{} discard queue", size, maxQueueSize);
        }
        if (size >= batchSize) {
            flush();
        }
    }


    public void flush() {
        Set<Long> deviceIds = new HashSet<>();
        while (!heartbeatQueue.isEmpty() && deviceIds.size() <= batchSize) {
            PingVO pingVO = heartbeatQueue.poll();
            // 没有了也批量一次
            if (null == pingVO) {
                break;
            }
            deviceIds.add(pingVO.getDeviceId());
        }
        if (!deviceIds.isEmpty()) {
            deviceInfoService.ping(deviceIds);
        }
    }
}
